package ru.unlocker.reports.domain;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public abstract class FoldingReportStreamJob {

  public final StreamExecutionEnvironment createApplicationPipeline(
      ParameterTool params, StreamExecutionEnvironment env
  ) {
    env.getConfig().setGlobalJobParameters(params);
    var reportStream = readReportStream(params, env);
    var producedStream = reportStream
        .keyBy(RawReportEvent::getReportId)
        .process(new ReportEventProcessor());
    writeReportStream(params, producedStream);
    return env;
  }

  protected abstract DataStream<RawReportEvent> readReportStream(ParameterTool params, StreamExecutionEnvironment env);

  protected abstract void writeReportStream(ParameterTool params, DataStream<ReportAndTimerEvent> outStream);

}
